package com.ebupt.migu.music.service.impl;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import cn.hutool.core.thread.NamedThreadFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.stereotype.Service;
import com.ebupt.migu.music.entity.UserGroupInfo;
import com.ebupt.migu.music.mapper.UserGroupInfoMapper;
import com.ebupt.migu.music.redis.service.RedisService;
import com.ebupt.migu.music.service.SyncTsgService;
import com.ebupt.migu.restapi.DownloadByFtp;
import com.ebupt.migu.restapi.GroupDetailDel;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * @author tz
 */
@Slf4j
@Service
public class SyncTsgServiceImpl implements SyncTsgService {

    @Value("${tsg.pro}")
    private String filePro;
    @Value("${tsg.path}")
    private String filePath;
    @Value("${tsg.post_url}")
    private String postUrl;
    @Resource
    private RedisService redisService;
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private UserGroupInfoMapper userGroupInfoMapper;

    @Override
    public void syncUserGroup() {
        DateTime start = DateUtil.beginOfDay(DateUtil.yesterday());
        DateTime end = DateUtil.endOfDay(DateUtil.yesterday());
        getCustomer(start, end);
    }

    @Override
    public void fileDataToRedis() {
        List<UserGroupInfo> infos = userGroupInfoMapper.qryCustomer();
        if (infos.size() > 0) {
            infos.forEach(e -> {
                try {
                    log.info("开始更新人群：{}", e.getGroupId() + "-" + e.getUserGroupName());
                    userGroupInfoMapper.updateUserGroupDataWrittenToRedis(e.getGroupId(), 2);
                    Map<String, Integer> conut = syncUserGroupUid(e.getGroupId());
                    userGroupInfoMapper.updateClientUser(e.getGroupId(),conut.get("clientUser"),conut.get("mobileUser"));
                } catch (Exception exception) {
                    exception.printStackTrace();
                    log.error("TSG客群数据录入Redis异常：{}", exception.getMessage());
                    userGroupInfoMapper.updateUserGroupDataWrittenToRedis(e.getGroupId(), 1);
                }
            });
        }
    }

    /**
     * 获取指定时段的客群
     *
     * @param start
     * @param end
     */
    public void getCustomer(DateTime start, DateTime end) {
        log.info("开始时间：{}，结束时间：{}", start, end);
        List<UserGroupInfo> infos = userGroupInfoMapper.syncTsgUserGroup(start, end);
        log.info("查询到符合条件的客群总计{}个", infos.size());
        String fileStr = filePro + DateUtil.format(new Date(), "yyyyMMdd");
        if (infos.size() > 0) {
            for (int i = 0; i < infos.size(); i++) {
                String redisKey = infos.get(i).getGroupId() + "-phone";
                log.info("开始导出客群-> 客群ID：{}，客群名称：{}", infos.get(i).getGroupId(), infos.get(i).getUserGroupName());
                Set<Object> set = new HashSet<>();
                if (1 == infos.get(i).getIsStaticCustomer()) {
                    log.info("开始获取TSG客群 === 人群id：{}，人群名称:{}", infos.get(i).getGroupId(),
                            infos.get(i).getUserGroupName());
                    Map<String, Object> strategy = new HashMap<>();
                    strategy.put("strategyId", System.currentTimeMillis());
                    strategy.put("strategyName", System.currentTimeMillis());
                    strategy.put("groupId", infos.get(i).getGroupId());
                    strategy.put("approverName", System.currentTimeMillis());
                    strategy.put("creatorName", System.currentTimeMillis());
                    File file = DownloadByFtp.getLabelUserGroup(strategy, postUrl, 1);
                    try (BufferedReader br = new BufferedReader(new FileReader(file))) {
                        String line;
                        while ((line = br.readLine()) != null) {
                            set.add(line);
                        }
                        log.info("客群规模-> 客群ID：{}，SIZE：{}", infos.get(i).getGroupId(), set.size());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    if (file.delete()) {
                        GroupDetailDel.del(strategy);
                    }
                } else {
                    log.info("Redis本地客群，从Redis中导出：{}", redisKey);
                    long msisdns = redisTemplate.opsForSet().size(redisKey);
                    if (msisdns > 20000000) {
                        log.info("客群规模大于2000万，无法导出-> KEY：{}，SIZE：{}", redisKey, msisdns);
                        continue;
                    }
                    set = redisService.getSet(redisKey);
                    log.info("客群规模-> KEY：{}，SIZE：{}", redisKey, set.size());
                }
                if (null == set || set.size() == 0) {
                    set.add(com.ebupt.migu.common.util.DateUtil.getDateToday());
                }
                // 生成数据文件
                String datafile = i < 10 ? fileStr + "_0" + i + "_001.dat" : fileStr + "_" + i + "_001.dat";
                File f1 = FileUtil.writeLines(set, filePath + datafile, "UTF-8");
                // 生成校验文件
                String vertFile = i < 10 ? fileStr + "_0" + i + ".verf" : fileStr + "_" + i + ".verf";
                String str = String.join("|^|", f1.getName(), String.valueOf(f1.length()), String.valueOf(set.size()),
                        infos.get(i).getUserGroupName(), infos.get(i).getGroupId(),
                        DateUtil.format(new Date(), "yyyyMMddHHmmss"));
                File f2 = FileUtil.writeString(str, filePath + vertFile, "UTF-8");
                log.info("人群名称：{},数据文件：{},校验文件：{}", infos.get(i).getUserGroupName(), f1.getName(), f2.getName());
            }
        } else {
            log.info("没有可同步tsg的人群信息");
        }
    }

    /**
     * 同步客群信息至redis
     *
     * @param groupId 客群id
     */
    private Map<String, Integer> syncUserGroupUid(String groupId) {
        // 设置3个月过期时间
        Long expire = 1000L * 60 * 60 * 24 * 90;
        log.info("开始更新人群：{}", groupId);
        Map<String, Object> strategy = new HashMap<>();
        strategy.put("strategyId", System.currentTimeMillis());
        strategy.put("strategyName", System.currentTimeMillis());
        strategy.put("groupId", groupId);
        strategy.put("approverName", System.currentTimeMillis());
        strategy.put("creatorName", System.currentTimeMillis());
        File phoneFile = DownloadByFtp.getLabelUserGroup(strategy, postUrl, 1);
        int mobileUser = getGroupInfoToRedis(phoneFile, groupId + "-phone", expire);
        File uidFile = DownloadByFtp.getLabelUserGroup(strategy, postUrl, 3);
        int clientUser = getGroupInfoToRedis(uidFile, groupId + "-uid", expire);
        Map<String, Integer> count = new HashMap<>();
        count.put("mobileUser", mobileUser);
        count.put("clientUser", clientUser);
        return count;
    }

    /**
     * 读取人群信息存入redis中
     *
     * @param file       文件信息
     * @param key        redis的key
     * @param expireTime 过期时间
     */
    private int getGroupInfoToRedis(File file, String key, Long expireTime) {
        if (file != null && file.length() != 0) {
            String str;
            Set<String> set = new HashSet<>();
            try {
                BufferedReader br = new BufferedReader(new FileReader(file));
                while ((str = br.readLine()) != null) {
                    if (!"".equals(str) && !str.startsWith("GEN_")) {
                        set.add(str);
                    }
                }
                setCollectToRedis(key, set, expireTime);
                br.close();
            } catch (IOException e) {
                log.info("读取标签平台文件异常：{}", e.getMessage());
            }
            return set.size();
        }
        return 0;
    }

    public void setCollectToRedis(final String key, Set<String> value, Long expireTime) {
        try {
            SetOperations opsForSet = redisTemplate.opsForSet();
            int size = value.size();
            if (size >= 1000000) {
                ThreadPoolExecutor executor = new ThreadPoolExecutor(5,
                        10,
                        5000,
                        TimeUnit.MILLISECONDS,
                        new ArrayBlockingQueue<>(5),
                        new ThreadPoolExecutor.AbortPolicy());
                executor.submit(() -> {
                    Set<String> set = new HashSet<>();
                    for (String next : value) {
                        set.add(next);
                        if (set.size() >= 100000) {
                            opsForSet.add(key, set.toArray());
                            set = new HashSet<>();
                            redisTemplate.expire(key, expireTime, TimeUnit.MILLISECONDS);
                        }
                    }
                    if (set.size() > 0) {
                        opsForSet.add(key, set.toArray());
                        redisTemplate.expire(key, expireTime, TimeUnit.MILLISECONDS);
                    }
                });
                executor.shutdown();
            } else {
                opsForSet.add(key, value.toArray());
                redisTemplate.expire(key, expireTime, TimeUnit.MILLISECONDS);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
